//#include "gateway_error.h"
//#include "gateway_packet.h"
#include "gateway_client.h"
#include <boost/regex.hpp>
#include <boost/format.hpp>
#include <boost/lexical_cast.hpp>  
#include <boost/make_shared.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/ini_parser.hpp>
#include <boost/property_tree/json_parser.hpp>
#include <string.h>

#include "../tool/crc16.h"
#include "../tool/encrpty.h"
#include "../tool/cbase64.h"
#include "../pagefile/pagefile.h"
#include "../mqttservice/mqttservice.h"
#include "../tool/md5.h"
#include "../../liblog/Include/liblog.h"
#include "gateway_taskpool.h"
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/date_time.hpp>  
#include "../mqttservice/mqttservice.h"
#include "gateway_cache.h"
#include <iostream>  
#include <sstream>
#include <boost/locale.hpp>

using namespace boost::property_tree;
using namespace boost::gregorian;
using namespace boost::posix_time;
using namespace boost;
using namespace std;

const int time_out = 300;// s

gateway_client::gateway_client(gateway_stream_ptr stream_ptr)
 :_is_connected(false)
 ,_ping_time(10) //20s
 ,_car_cnt_time(1) //2分钟
 ,_other_cnt_time(1) //3分钟
 ,_isn(0)
 ,_strhid("")
 ,_stream_ptr(stream_ptr)
 ,_mqtt(NULL) {
}

gateway_client::~gateway_client() {
    if (_mqtt)
        delete _mqtt;
    _mqtt = NULL;
    //printf("%ld New client[%s:%u] ~gateway_client().\n", time(NULL), _host.c_str(), _port);
}

/**** gateway_stream_event ****/
int gateway_client::stream_init() {
	_host = _stream_ptr->get_remote_host();
    _port = _stream_ptr->get_remote_port();
    _addr = _host + std::string(":") +boost::lexical_cast<std::string>(_port);
    get_mqtt_config(_mqttconfig);
    _mqtt = new CMQTTService(this);
    boost::property_tree::ptree reader;
    boost::property_tree::ini_parser::read_ini("config.ini", reader);
	//server_ip = reader.get<std::string>("collection.SERVER_IP");
	_ping_time = reader.get<unsigned short>("time.PINT_TIME");
    _car_cnt_time = reader.get<unsigned char>("time.CAR_CNT_TIME");
    _other_cnt_time = reader.get<unsigned char>("time.CAR_CNT_TIME");
    _stream_ptr->set_timeout(_ping_time);
	//printf("%ld New client[%s] connected.\n", time(NULL), _addr.c_str());
    _client_ptr.reset(this);
	return 0;
}

int gateway_client::stream_clean() {
	//printf("%ld Client[%s:%u] disconnected.\n", time(NULL), _host.c_str(), _port);
    if (_mqtt)
    {
        _mqtt->Stop();
        delete _mqtt;
        _mqtt = NULL;
    }
    gateway_cache::get_instance()->del_client(_strhid, _addr);
	_client_ptr.reset();
	return 0;
}

int gateway_client::stream_read(unsigned char* data, unsigned int dataLen) {
    unsigned int used = 0;
    _stream_ptr->set_timeout(30);//超时30s
	int ret = handle_parse_packet(data, dataLen, used);
	if (ret < 0)
	{
		//log
		_stream_ptr->close();
		return -1;
	}
	return std::min(dataLen, used);
}

/**** gateway_task ****/
int gateway_client::on_task(void* param) {
	if (param == NULL)
		return -1;
    if (_mqtt == NULL)
        return -1;
    event_data *ped = (event_data*)param;
    if (ped->tasktype == PULISH)
    {
        int ret = 0;
        pulish_data *ppdata = (pulish_data*)(ped->data);
        std::string strjson = std::string(ppdata->data, ppdata->datalen);
        delete[] ppdata->data;
        delete ppdata;
        ret = _mqtt->pushUpdateData(strjson.length(), (const void*)strjson.c_str());
        if (ret)
        {
            //log
            LogError(("%s-mqttpush ret:%d, val:%s\r\n", __PRETTY_FUNCTION__, ret, strjson.c_str()));            
        }
    }
    else if (ped->tasktype == SUBSCRIBE)
    {
        onmessage_data *pmsg = (onmessage_data*)(ped->data);
        task_onmessage(pmsg->strid, pmsg->actions, pmsg->payload);
        delete pmsg;
    }
    else if (ped->tasktype == ONCONNECT)
    {
        _mqtt->devSubscribe();
        //执行get操作
        //_mqtt->pushGetData();
    }
    delete ped; 
    return 0;
}

int gateway_client::handle_parse_packet(unsigned char* buf, unsigned int bufLen, unsigned int &used) {
    unsigned char * ptmp = buf;
    used = 0;
    int ret = 0;
    LogDebug(("RCV:"));
    //display_buf_byhex(buf, bufLen);
    //unsigned int pos = 0; 
    do{
        for (;ptmp[used] != 0xc0; used++)
        {
            if (used >= bufLen)
                return  -1;
        }  
        //used = pos;
        unsigned int total = bufLen - used;
        std::string protoText((char*)(buf + used), total);
        display_buf_byhex((const unsigned char*)protoText.c_str(), protoText.length());
        switch((unsigned char)protoText.at(3))
        {
        case POWERON:
        {
            ret = deal_poweron(protoText, used);
            if (ret)
            {
                LogError(("%s:处理屏幕上电数据错误ret=%d\r\n", _strhid.c_str(),ret));
                //log;
            }
            break;
        }
        case PING:
        {
            deal_ping(protoText, used);
            break;
        }
        case CARVOL:
        {
            ret = deal_car_vol(protoText, used);
            if (ret)
            {
                LogError(("%s:处理车流量错误ret=%d\r\n", _strhid.c_str(),ret));
                //log
            }
            break;
        }
        case SCREENDATA:
        {
            ret = deal_screen_data(protoText, used);
            if (ret)
            {
                LogError(("%s:处理屏幕据错误ret=%d\r\n", _strhid.c_str(),ret));
                //log
            }
            break;
        }
        case DICIDATA:
        {
            ret = deal_dici_data(protoText, used);
            if (ret)
            {
                LogError(("%s:处理地磁数据错误ret=%d\r\n", _strhid.c_str(),ret));
                //log
            }
            break;
        }
        default:
        {
            //log
            LogError(("%s:不能识别的数据帧类型:%02X\r\n", _strhid.c_str(), protoText.at(used + 3)));
            return -1;
        }
        };
        if(ret == 1)
            return 0;
        else if(ret < 0)
            return ret;
    }while((ret == 0) && (used < bufLen));
    return ret;
}

void gateway_client::display_buf_byhex(const unsigned char * buf, unsigned int len)
{
    if (!buf)
        return;
    for (unsigned int i = 0; i <len; i++)
    {
        printf("%02X ", buf[i]);
    }
    printf("\r\n");
}

unsigned int gateway_client::get_screenmac(const unsigned char* buf, int buflen)
{
    if (!buf)
        return 0;

    if (buflen < 4)
        return 0;
    
    unsigned int id = 0;
    id = buf[0];
    id = id << 8 | buf[1];
    id = id << 8 | buf[2];
    id = id << 8 | buf[3];
    return id;
}

std::string gateway_client::get_reporttime(const unsigned char* buf, int buflen)
{
    if (!buf)
        return "";
        
    if (buflen < 4)
        return "";
    time_t  utime = 0;
    utime = buf[0];
    utime = utime << 8 | buf[1];
    utime = utime << 8 | buf[2];
    utime = utime << 8 | buf[3];
    struct tm *p; 
    p = localtime(&utime);
    char s[100] = {0};
    strftime(s, sizeof(s), "%Y-%m-%d %H:%M:%S", p);  
    return std::string(s);
}

int gateway_client::get_carvol(const unsigned char* buf, int buflen)
{
    if (!buf)
        return -1;
        
    if (buflen < 4)
        return -1;

    unsigned int  carcnt = 0;
    carcnt = buf[0];
    carcnt = carcnt << 8 | buf[1];
    carcnt = carcnt << 8 | buf[2];
    carcnt = carcnt << 8 | buf[3];
    return (int)carcnt;
}

unsigned short gateway_client::get_crc16(const unsigned char * buf, int buflen)
{
    if (!buf)
        return 0;
    crc16 crc;
    return crc.ModBus_CRC16(buf, (unsigned int)buflen);
}

int gateway_client::get_rsp_data(byteAttr byteattr, std::string &strdata)
{
    std::string strrsp = "";
    strrsp.push_back(0xc0);
    byteattr.sn = ((byteattr.sn & 0xf0) | _isn++);
    if (_isn > 0xf)
        _isn = 0;
    strrsp.push_back(byteattr.sn);
    strrsp.push_back(0x81);//通信参数，以太网通讯，服务器使用以太网与外界通讯
    strrsp.push_back(0x32);//功能标志，不需要回复，服务器回复
    strrsp.push_back(0x0e);
    strrsp.push_back(byteattr.srcAddr);
    strrsp.push_back(0x0c); //后面数据长度，crc16校验值除外
    unsigned int timetmp = (unsigned int)get_current_stamp();
    strrsp.push_back((timetmp >> 24) & 0xff);
    strrsp.push_back((timetmp >> 16) & 0xff);
    strrsp.push_back((timetmp >> 8) & 0xff);
    strrsp.push_back(timetmp & 0xff);

    strrsp.push_back((byteattr.screenMac >> 24) & 0xff);
    strrsp.push_back((byteattr.screenMac >> 16) & 0xff);
    strrsp.push_back((byteattr.screenMac >> 8) & 0xff);
    strrsp.push_back(byteattr.screenMac & 0xff);

    strrsp.push_back((_ping_time >> 8) & 0xff);
    strrsp.push_back(_ping_time & 0xff);

    strrsp.push_back(_car_cnt_time);
    strrsp.push_back(_other_cnt_time);

    unsigned short crc16 = get_crc16((const unsigned char*)strrsp.c_str(), strrsp.length());
    strrsp.push_back((crc16 >> 8) & 0xff);
    strrsp.push_back(crc16 & 0xff);
    strdata = strrsp;
    //LogDebug(("ACK:"));
    //display_buf_byhex((const unsigned char*)strdata.c_str(), strdata.length());
    return 0;
}

std::string gateway_client::get_poweron_json(powerOn poweron)
{
    std::string strjsonformat = "{\"hardwareid\":\"%s\",\"sc%s_commparm\":\"%s\",\"reporttime\":\"%s\"}";
    boost::format fmt(strjsonformat.c_str());
    fmt %poweron.screenmac % poweron.screenid % poweron.commparm %poweron.reporttime;
    std::string strjsondata = fmt.str();
    return strjsondata;
}

std::string gateway_client::get_car_vol(carVol carvol)
{
    std::string strjsonformat = "{\"hardwareid\":\"%s\",\"sc%s_To_sc%s\":\"%s\",\"sc%s_To_sc%s\":\"%s\",\"reporttime\":\"%s\"}";
    boost::format fmt(strjsonformat.c_str());
    fmt %carvol.screenmac % carvol.localscid % carvol.remotescid % carvol.localTrafficVol %  carvol.remotescid % carvol.localscid % carvol.remoteTrafficVol % carvol.reporttime;
    std::string strjsondata = fmt.str();
    return strjsondata;
}

std::string gateway_client::get_screen_data(screenData screendata)
{
    std::string strjsonformat = "{\"hardwareid\":\"%s\",\"sc%s_setNum\":\"%s\",\"sc%s_mac\":\"%s\",\"sc%s_voltage\":\"%s\",\"sc%s_sigNoise\":\"%s\",\"sc%s_sigStrength\":\"%s\","
    "\"sc%s_WLparm\":\"%s\",\"sc%s_diciDist\":\"%s\",\"sc%s_spdLmt\":\"%s\",\"sc%s_localDelay\":\"%s\",\"sc%s_remoteDelay\":\"%s\",\"sc%s_remoteCarDisplay\":\"%s\","
    "\"sc%s_localDisplay\":\"%s\",\"sc%s_display\":\"%s\",\"reporttime\":\"%s\"}";
    boost::format fmt(strjsonformat.c_str());
    fmt %screendata.screenmac 
    %screendata.screenId %screendata.setNum
    %screendata.screenId %screendata.scMac
    %screendata.screenId %screendata.screenVoltage
    %screendata.screenId %screendata.sigNoise
    %screendata.screenId %screendata.sigStrength
    %screendata.screenId %screendata.WLParm
    %screendata.screenId %screendata.diciDist
    %screendata.screenId %screendata.spdlmt
    %screendata.screenId %screendata.localdelay
    %screendata.screenId %screendata.remoteDelay
    %screendata.screenId %screendata.remoteCarDisplay
    %screendata.screenId %screendata.localDisplay
    %screendata.screenId %screendata.display
    %screendata.reporttime;
    std::string strjsondata = fmt.str();
    return strjsondata;
}

std::string gateway_client::get_dici_data(diciData dicidata)
{
    std::string strjsonformat = "{\"hardwareid\":\"%s\",\"dici%s_mac\":\"%s\",\"dici%s_voltage\":\"%s\",\"dici%s_sigNoise\":\"%s\",\"dici%s_sigStrength\":\"%s\",\"reporttime\":\"%s\"}";
    boost::format fmt(strjsonformat.c_str());
    fmt %dicidata.screenmac % dicidata.diciId %dicidata.diciMac
    % dicidata.diciId %dicidata.diciVoltage
    % dicidata.diciId %dicidata.sigNoise
    % dicidata.diciId %dicidata.sigStrength
    % dicidata.reporttime;
    std::string strjsondata = fmt.str();
    return strjsondata;
}

int gateway_client::deal_poweron(std::string strdata, unsigned int &used)
{
    const unsigned char * ptmp = (const unsigned char * )strdata.c_str();
    if (ptmp[0] != 0xc0)
        return -1;
    if (ptmp[3]  != POWERON)
        return -1;
    if (strdata.length() < 17)//上电帧的长度
        return 1;

    byteAttr byteattr;
    byteattr.screenMac = get_screenmac(ptmp+11, strdata.length() - 11);
    byteattr.sn = ptmp[1];
    byteattr.srcAddr = ptmp[4];
    powerOn poweron;
    poweron.screenmac = int2str((int)byteattr.screenMac);
    poweron.screenid =byte2str(byteattr.srcAddr);
    poweron.reporttime = get_reporttime(ptmp + 7, strdata.length() - 7);
    poweron.commparm = byte2str(ptmp[2]);
    if (_strhid.compare(std::string("")) ==0)
    {
        _strhid = poweron.screenmac;
    }
    else if (_strhid.compare(std::string(poweron.screenmac)) != 0)
    {
        //log 数据id不一样
        return -1;
    }

    //响应下发
    std::string strrspdata = "";
    get_rsp_data(byteattr, strrspdata);
    int ret = _stream_ptr->send((unsigned char*)strrspdata.c_str(), strrspdata.length());
    if (ret != (int)strrspdata.length())
        return -1; //不成功立即返回

    std::string strjson =  get_poweron_json(poweron);
    LogDebug(("powerOn:%s\r\n", strjson.c_str()));            
    handle_mqtt_event(strjson);
    used += 17;
    return 0; 
}

int gateway_client::deal_ping(std::string strdata, unsigned int &used)
{
    const unsigned char * ptmp = (const unsigned char * )strdata.c_str();
    if (ptmp[0] != 0xc0)
        return -1;
    if (ptmp[3]  != PING)
        return -1;
    if (strdata.length() < 17)//心跳帧的长度
        return 1;

    ByteAttr byteattr;
    byteattr.screenMac = get_screenmac(ptmp+11, strdata.length() - 11);
    byteattr.sn = ptmp[1];
    byteattr.srcAddr = ptmp[4];
    LogDebug(("%d心跳 sn:%02x\r\n", byteattr.screenMac, byteattr.sn));
    //响应下发
    std::string strrspdata = "";
    get_rsp_data(byteattr, strrspdata);
    int ret = _stream_ptr->send((unsigned char*)strrspdata.c_str(), strrspdata.length());
    if (ret != (int)strrspdata.length())
        return -1; //不成功立即返回
    used += 17;
    return 0;
}

int gateway_client::deal_car_vol(std::string strdata, unsigned int &used)
{
    const unsigned char * ptmp = (const unsigned char * )strdata.c_str();
    if (ptmp[0] != 0xc0)
        return -1;
    if (ptmp[3]  != CARVOL)
        return -1;
    const int carVolDataLen = 25;
    if (strdata.length() < carVolDataLen)//帧的长度
        return 1;

    ByteAttr byteattr;
    byteattr.screenMac = get_screenmac(ptmp+11, strdata.length() - 11);
    byteattr.sn = ptmp[1];
    byteattr.srcAddr = ptmp[4];

    carVol carvol;
    carvol.screenmac = dword2str(byteattr.screenMac);
    carvol.reporttime = get_reporttime(ptmp + 7, strdata.length() - 7);
    if (byteattr.srcAddr == 0x09)//屏幕id要么是0x09,要么是0x0d;
    {
        carvol.localscid = "9";
        carvol.remotescid = "13";
    }
    else
    {
        carvol.localscid = "13";
        carvol.remotescid = "9";
    }
    int carcnt = get_carvol(ptmp + 15, strdata.length() - 15);
    if (carcnt == -1)
        return -1;
    carvol.localTrafficVol = boost::lexical_cast<std::string>(carcnt);
    carcnt = get_carvol(ptmp + 19, strdata.length() - 19);
    if (carcnt == -1)
        return -1;
    carvol.remoteTrafficVol = boost::lexical_cast<std::string>(carcnt);

    if (_strhid.compare(std::string("")) ==0)
    {
        _strhid = carvol.screenmac;
    }
    else if (_strhid.compare(std::string(carvol.screenmac)) != 0)
    {
        //log 数据id不一样
        return -1;
    }

    //响应下发
    std::string strrspdata = "";
    get_rsp_data(byteattr, strrspdata);
    int ret = _stream_ptr->send((unsigned char*)strrspdata.c_str(), strrspdata.length());
    if (ret != (int)strrspdata.length())
        return -1; //不成功立即返回

    std::string strjson =  get_car_vol(carvol);
    //LogDebug(("carVol:%s\r\n", strjson.c_str()));      
    handle_mqtt_event(strjson);
    used += carVolDataLen;
    return 0;
}

int gateway_client::deal_screen_data(std::string strdata, unsigned int &used)
{
    const unsigned char * ptmp = (const unsigned char * )strdata.c_str();
    if (ptmp[0] != 0xc0)
        return -1;
    if (ptmp[3]  != SCREENDATA)
        return -1;
    const int screenDataLen = 89;
    if (strdata.length() < screenDataLen)//帧的长度
        return 1;

    ByteAttr byteattr;
    byteattr.screenMac = get_screenmac(ptmp+11, strdata.length() - 11);
    byteattr.sn = ptmp[1];
    byteattr.srcAddr = ptmp[4];
    
    screenData screendata;
    screendata.screenmac = dword2str(byteattr.screenMac);
    screendata.reporttime = get_reporttime(ptmp + 7, strdata.length() - 7);

    unsigned int scmac = get_screenmac(ptmp+15, strdata.length() - 15);
    screendata.scMac = dword2str(scmac);
    
    int pos = 19;
    unsigned short scsetnum = ptmp[pos++];
    scsetnum = (scsetnum << 8| ptmp[pos++]);
    screendata.setNum = word2str(scsetnum);
    screendata.screenId = byte2str(ptmp[pos++]);
    float valtage = 7.5 + (ptmp[pos++] * 0.1);
    screendata.screenVoltage = boost::lexical_cast<std::string>(valtage);
    screendata.sigNoise = byte2str(ptmp[pos++]);
    screendata.sigStrength = byte2str(ptmp[pos++]);
    screendata.WLParm = byte2str(ptmp[pos++]);
    screendata.diciDist = byte2str(ptmp[pos++]);
    screendata.spdlmt = byte2str(ptmp[pos++]);
    screendata.localdelay = byte2str(ptmp[pos++]);
    screendata.remoteDelay = byte2str(ptmp[pos++]);
    screendata.remoteCarDisplay = std::string((const char*)(ptmp+pos), 8);//是GB2312格式要转换成utf-8格式
    screendata.remoteCarDisplay = boost::locale::conv::between( screendata.remoteCarDisplay, "UTF-8", "GB2312" );  
    pos +=8;
    screendata.localDisplay = std::string((const char*)(ptmp+pos), 8);
    screendata.localDisplay = boost::locale::conv::between( screendata.localDisplay, "UTF-8", "GB2312" );  
    pos +=8;
    int len = ptmp[pos++];
    len = (len > 40) ? 40 : len;
    screendata.display = std::string((char*)(ptmp + pos), len);
    screendata.display = boost::locale::conv::between( screendata.display, "UTF-8", "GB2312" );  
    if (_strhid.compare(std::string("")) ==0)
    {
        _strhid = screendata.screenmac;
    }
    else if (_strhid.compare(std::string(screendata.screenmac)) != 0)
    {
        //log 数据id不一样
        return -1;
    }

     //响应下发
     std::string strrspdata = "";
     get_rsp_data(byteattr, strrspdata);
     int ret = _stream_ptr->send((unsigned char*)strrspdata.c_str(), strrspdata.length());
     if (ret != (int)strrspdata.length())
         return -1; //不成功立即返回
 
     std::string strjson =  get_screen_data(screendata);
     //LogDebug(("screenData:%s\r\n", strjson.c_str()));      
     handle_mqtt_event(strjson);
     used += screenDataLen;
     return 0;
}

int gateway_client::deal_dici_data(std::string strdata, unsigned int &used)
{
    const unsigned char * ptmp = (const unsigned char * )strdata.c_str();
    if (ptmp[0] != 0xc0)
        return -1;
    if (ptmp[3]  != DICIDATA)
        return -1;
    if (strdata.at(5) != 0x0e)//标识是上报给服务器的帧
        return -1;

    const int diciDataLen = 25;
    if (strdata.length() < diciDataLen)//帧的长度
        return 1;

    ByteAttr byteattr;
    byteattr.screenMac = get_screenmac(ptmp+11, strdata.length() - 11);
    byteattr.sn = ptmp[1];
    byteattr.srcAddr = ptmp[4];

    diciData dicidata;
    dicidata.screenmac = dword2str(byteattr.screenMac);
    dicidata.reporttime = get_reporttime(ptmp + 7, strdata.length() - 7);
    unsigned int dicimac = get_screenmac(ptmp+15, strdata.length() - 15);
    dicidata.diciMac = dword2str(dicimac);
    dicidata.diciId =  byte2str(ptmp[19]);

    float valtage = 2.5 + (ptmp[20] * 0.02);
    dicidata.diciVoltage = boost::lexical_cast<std::string>(valtage);
    dicidata.sigNoise = byte2str(ptmp[21]);
    dicidata.sigStrength = byte2str(ptmp[22]);
    
    if (_strhid.compare(std::string("")) ==0)
    {
        _strhid = dicidata.screenmac;
    }
    else if (_strhid.compare(std::string(dicidata.screenmac)) != 0)
    {
        //log 数据id不一样
        return -1;
    }

     //响应下发
     std::string strrspdata = "";
     get_rsp_data(byteattr, strrspdata);
     int ret = _stream_ptr->send((unsigned char*)strrspdata.c_str(), strrspdata.length());
     if (ret != (int)strrspdata.length())
         return -1; //不成功立即返回
 
     std::string strjson =  get_dici_data(dicidata);
     //LogDebug(("diciData:%s\r\n", strjson.c_str()));      
     handle_mqtt_event(strjson);
     used += diciDataLen;
     return 0;

}

int gateway_client::handle_mqtt_event(std::string strjson)
{
	//获取mqtt id和key
	if (_mqtt->IsConnect() == false &&  _is_connected == false)//没有连接
	{
		std::string strdeviceid = "";
		std::string strdevicekey = "";
		if (!CPagefile::GetInstance().GetNewDeviceInfo(_strhid, strdeviceid, strdevicekey))
			return -1;

		MD5 md5(strdeviceid + strdevicekey);
		_mqttconfig.strusrname = strdeviceid;
		_mqttconfig.strmqttid = strdeviceid;
		_mqttconfig.strusrpwd = md5.hexdigest();
        _mqttconfig.strkey = strdevicekey;
        
        //
        LogDebug(("%s:mqttid- %s\r\n",  _strhid.c_str(), _mqttconfig.strmqttid.c_str()));
        //
        gateway_cache::get_instance()->insert_client(_strhid, _client_ptr);
		if (_mqtt->Start(_mqttconfig))
			return -1;		

		//if (_mqtt->devSubscribe())
		//{
		//	_mqtt->disconnect();
		//	return -1;		
        //};
        
        _is_connected = true;
	}
    event_data *ed = new event_data;
    ed->tasktype = PULISH;
    pulish_data *pdata = new pulish_data;
    pdata->data = new char[strjson.length()];
    strjson.copy(pdata->data, strjson.length(), 0);
    pdata->datalen = strjson.length();
    ed->data = (void*) pdata;
    gateway_taskpool::get_instance()->schedule(boost::static_pointer_cast<gateway_task>(_client_ptr), (void*)ed);
    return 0;
}

int gateway_client::gateway_packet_ack(){
    return 0;
}

int gateway_client::get_mqtt_config(MQTTCONFIG & mqttconfig)
{
    
	std::string zk_hosts;
	try
	{
		boost::property_tree::ptree reader;
		boost::property_tree::ini_parser::read_ini("config.ini", reader);
		mqttconfig.strmqtthost = reader.get<std::string>("mqtt.HOSTS");
        mqttconfig.mqttport = reader.get<int>("mqtt.PORT");
	}
	catch (std::exception& ec)
	{
		mqttconfig.strmqtthost = "127.0.0.1";
		mqttconfig.mqttport = 1883;
		LogError(("%s failed.error:%s.\r\n", __PRETTY_FUNCTION__, ec.what()));
		return -1;
	}
    return 0;
}

/*
boost::shared_ptr<gateway_client> gateway_client::get_ptr()
{
    //return boost::dynamic_pointer_cast<gateway_client> (shared_from_this());
    return shared_from_this();
}
*/


int gateway_client::OnConnect(CMQTTService *mq, std::string strid, int rc)
{
    if (rc != 0)
    {
        //连接失败
		LogError(("%s:%s连接MQTT服务器失败,返回错误码:%d\r\n", __PRETTY_FUNCTION__, strid.c_str(), rc));
        return rc;
    }
    
    LogDebug(("%s 连接mqtt服务器成功\r\n", strid.c_str()));
    event_data *ped = new event_data;
    ped->tasktype =  ONCONNECT;
    ped->data = NULL;
    gateway_taskpool::get_instance()->schedule(boost::static_pointer_cast<gateway_task>(_client_ptr), (void*)ped);
    ///mq->devSubscribe();

    //执行get操作
    //mq->pushGetData();
    return 0;
}
int gateway_client::OnDisconnect(CMQTTService *mq, std::string strid, int rc)
{
	LogDebug(("%s:断开,返回码%d \r\n", strid.c_str(), rc));
    if (!mq)
        return 0;
    //mq->reconnect();
    
    return 0;
}
int gateway_client::OnMessage(CMQTTService *mq, std::string strid, ShadowActions actions, std::string payload)
{
	//if (!_stop)
    //_threadpool.schedule(boost::bind(&CTransaction::DealCommandInfo, this, strid, actions, payload));
    event_data *ed = new event_data;
    ed->tasktype = SUBSCRIBE;
    onmessage_data *pmsgdata = new onmessage_data;
    pmsgdata->strid = strid;
    pmsgdata->actions = actions;
    pmsgdata->payload = payload;
    ed->data = (void*)pmsgdata; 
    gateway_taskpool::get_instance()->schedule(boost::static_pointer_cast<gateway_task>(_client_ptr), (void*)ed);
    //LogDebug(("%s 收到消息%s\r\n", strid.c_str(), payload.c_str()));
    return 0;
}
int gateway_client::OnSubscribe(CMQTTService *mq, std::string strid, int mid, int qos_count, const int *granted_qos)
{
    LogDebug(("%s 订阅topic成功成功\r\n", strid.c_str()));
    return 0;
}

int gateway_client::OnUnsubscribe(CMQTTService *mq, std::string strid, int mid)
{
    return 0;
}

int gateway_client::OnPublish(CMQTTService *mq, std::string strid, int mid)
{
    LogDebug(("%s: 上报成功\r\n", strid.c_str()));
    return 0;
}


void gateway_client::close()
{
    _stream_ptr->close();
}

int gateway_client::task_pulish(std::string strpulishdata)
{
    int ret = 0;
    if (_mqtt == NULL)
        return -1;

    if (strpulishdata.length() <= 0)
        return -1;

    ret = _mqtt->pushUpdateData(strpulishdata.length(), (const void*)strpulishdata.c_str());
    if (ret)
    {
        //log
        LogError(("%s-mqttpush ret:%d, val:%s\r\n", __PRETTY_FUNCTION__, ret, strpulishdata.c_str()));
    }
    return ret;
}


void gateway_client::task_onmessage(std::string strid, ShadowActions actions, std::string strpayload)
{
    int ret = 0;
    switch (actions)
    {
    case SHADOW_UPDATE:
    {
        if ((ret = deal_update(strpayload)) != 0)
            LogError(("%s: 设备更新影子失败 错误%d\r\n", strid.c_str(), ret));
        break;
    }
    /*
    case SHADOW_GET:
    {
        if (deal_get(strid, strpayload) != 0)
            LogError(("%s: 设备主动获取影子数据%s处理失败\r\n", strid.c_str(), strpayload.c_str()));
        break;
    }
    case SHADOW_CTRL:
    {
        if (deal_ctrl(strid, strpayload) != 0)
            LogError(("%s: 设备控制命令下发处理失败\r\n", strid.c_str()));
        break;
    }
    */
    default:
        return;
    }
    return;

}

std::string gateway_client::get_json_val(const std::string strJson, const std::string strkey)
{
    std::stringstream stream(strJson);
    boost::property_tree::ptree pt;
    std::string strjsonval = "";
    try{  
        read_json<boost::property_tree::ptree>(stream, pt);
        strjsonval = pt.get<std::string>(strkey); 
    }catch(boost::property_tree::ptree_error & e) {  
        LogError(("%s: %s json form error\r\n", __PRETTY_FUNCTION__, strJson.c_str()));
        strjsonval = "";   
    }
    return strjsonval;

}

int gateway_client::deal_update(const std::string strJson)
{
    std::string strstatus = get_json_val(strJson, "status");
    if (strstatus.length() == 0)
        return -1;

    int istat = boost::lexical_cast<int>(strstatus);
    if (istat != 200)
    {
        return istat;
    }
    return 0;
}

int gateway_client::get_current_stamp()
{
    ptime epoch(boost::gregorian::date(1970, boost::gregorian::Jan, 1));
    time_duration time_from_epoch = second_clock::universal_time() - epoch;
    return time_from_epoch.total_seconds();
} 

std::string gateway_client::byte2str(const unsigned char srcbyte, int decimal)
{
    char s[4]={0,0,0,0};           //设定12位对于存储32位int值足够
    if (decimal == 10)
        snprintf(s,4, "%u", srcbyte);
    else
        snprintf(s,4 ,"%02x", srcbyte);
    string str = s;
    return str;
}

std::string gateway_client::word2str(const unsigned short srcword, int decimal)
{
    char s[4]={0,0,0,0};           //设定12位对于存储32位int值足够
    if (decimal == 10)
        snprintf(s,4, "%u", srcword);
    else
        snprintf(s,4 ,"%02x", srcword);
    string str = s;
    return str;
}

std::string gateway_client::dword2str(const unsigned int srcdword, int decimal)
{
    char s[4]={0,0,0,0};           //设定12位对于存储32位int值足够
    if (decimal == 10)
        snprintf(s,4, "%u", srcdword);
    else
        snprintf(s,4 ,"%02x", srcdword);
    string str = s;
    return str;
}

std::string gateway_client::int2str(const int srcint,int decimal)
{
    char s[12]={0,0,0,0,0,0,0,0,0,0,0,0};           //设定12位对于存储32位int值足够
    if (decimal == 10)
        snprintf(s,12, "%u", srcint);
    else
        snprintf(s,12 ,"%02x", srcint);
    string str = s;
    return str;
}